package com.heima.kafka.sample;

import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.KeyValue;
import org.apache.kafka.streams.StreamsBuilder;
import org.apache.kafka.streams.StreamsConfig;
import org.apache.kafka.streams.kstream.KStream;
import org.apache.kafka.streams.kstream.TimeWindows;
import org.apache.kafka.streams.kstream.ValueMapper;

import java.time.Duration;
import java.util.Arrays;
import java.util.Properties;

/**
 * 流式处理
 */
public class KafkaStreamQuickStart {

    public static void main(String[] args) {
        // kafkastream的配置信息
        Properties prop=new Properties();
        prop.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,"192.168.200.130:9092");
        prop.put(StreamsConfig.APPLICATION_ID_CONFIG,"streams-quickstart");
        prop.put(StreamsConfig.DEFAULT_KEY_SERDE_CLASS_CONFIG, Serdes.String().getClass());
        prop.put(StreamsConfig.DEFAULT_VALUE_SERDE_CLASS_CONFIG, Serdes.String().getClass());

        // stream 拓扑构建器
        final StreamsBuilder streamsBuilder=new StreamsBuilder();

        // 流式计算
        streamProcessor(streamsBuilder);

        // 构建kafkaStream对象
        KafkaStreams kafkaStreams=new KafkaStreams(streamsBuilder.build(),prop);

        //开启流式计算
        kafkaStreams.start();
    }


    /**
     * 流式计算
     * 消息的内容：hello kafka
     * @param streamsBuilder
     */
    private static void streamProcessor(StreamsBuilder streamsBuilder) {
        // 拓扑构建器从名为 Kafka 的主题创建源流：quickstart-topic-input 从指定的主题接收消息
        //创建KStream对象，同时指定从那个topic中接收消息
        KStream<String, String> source = streamsBuilder.stream("quickstart-topic-input");

       source.flatMapValues(new ValueMapper<String, Iterable<?>>() {
           // 拆分消息
           @Override
           public Iterable<?> apply(String value) {
               return Arrays.asList(value.split(" "));
           }
       })
               //按照value进行聚合处理
               .groupBy((key,value)->value)
               //时间窗口
               .windowedBy(TimeWindows.of(Duration.ofSeconds(10)))
               //统计单词的个数
               .count()
               //转换为kStream
               .toStream()
               .map((key,value)->{
                   System.out.println("key:"+key+",value:"+value);
                   return new KeyValue<>(key.key().toString(),value.toString());
               })
               //发送消息
               .to("quickstart-topic-output");


        // 现在，我们得到了一个从其源 Kafka 主题持续生成记录的 . 记录被组织为键入的键值对。
        // 我们可以对这个流做的最简单的事情是将其写入另一个 Kafka 主题，
        // 比如说它被命名为：KStream quickstart-topic-input  String quickstart-topic-output
//        source.to("quickstart-topic-output");

        // 上面等同于 streamsBuilder.stream("quickstart-topic-input").to("quickstart-topic-output");

    }
}
